package com.sdg.offline

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.mllib.recommendation.{ALS, Rating}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.jblas.DoubleMatrix


/**
  * 离线推荐
  * 1.训练模型
  * 2.计算用户推荐矩阵
  * 3.电影相似度矩阵
  */
object OfflineRecommender {

  val MONGO_URI: String = "mongodb://vm2:27017/recom3"

  val MONGODB_DATABASE: String = "recom3"

  val MONGODB_RATING_COLLECTION = "Rating"

  val MONGO_DRVIVE_CLASS: String = "com.mongodb.spark.sql"

  val MONGODB_MOVIE_COLLECTION = "Movie"

  val USER_MAX_RECOMMENDATION: Int = 20


  val MONGODB_USER_RECS_COLLECTION = "UserRecs"

  val MONGODB_MOVIE_RECS_COLLECTION = "MoviesRecs"

  /**
    * 计算两个电影之间的余弦相似度
    *
    * @param movie1
    * @param movie2
    */
  def consinSim(movie1: DoubleMatrix, movie2: DoubleMatrix): Double = {
    //dot是点乘
    //norm2()  是向量取模
    movie1.dot(movie2) / (movie1.norm2() * movie2.norm2())
  }


  def main(args: Array[String]): Unit = {
    //封装配置信息
    val config = Map(
      "spark.cores" -> "local",
      "mongo.uri" -> MONGO_URI,
      "mongo.db" -> MONGODB_DATABASE
    )



    //conf对象
    val sparkConf = new SparkConf().setAppName("OfflineRecommender").setMaster(config("spark.cores"))
      //在程序中设置资源参数
      .set("spark.executor.memory", "6G").set("spark.driver.memory", "3G")



    //SparkSession
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    //读取mongoDb 的数据
    val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))
    import spark.implicits._
    //从mongo的表中去读数据 [评分数据表] 数据格式 [用户id,电影id,评分]
    //todo:1.ratingRDD
    val ratingRDD = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .format(MONGO_DRVIVE_CLASS)
      .load()
      //[用户id,电影的id,用户对电影的评分,评分的时间]
      .as[MoviesRating]
      .rdd
      //[用户id,电影的id,用户对电影的评分]
      .map(rating => (rating.uid, rating.mid, rating.score))
      //如果不增加缓存的话,就会出现内存溢出的问题
      .cache()


    //去重操作用户的数据集 RDD[Int] [用户id]
    //todo:2.userRDD
    val userRDD = ratingRDD.map(_._1).distinct()

    //电影数据集 RDD[Int]
    //todo:3.movieRDD
    val movieRDD = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_MOVIE_COLLECTION)
      .format(MONGO_DRVIVE_CLASS)
      .load()
      //对应的电影.下面才能制定字段去数据
      .as[Movie]
      .rdd
      .map(_.mid)
      //如果不增加缓存的话,就会出现内存溢出的问题
      .cache()

    //创建训练数据集 [userid,mid,soucre]
    val trainData = ratingRDD.map(x => Rating(x._1, x._2, x._3))
    //特征数   迭代次数  正则参数
    val (rank, iterations, lambda) = (10, 1, 0.01)
    //模型训练[训练集,特征数,迭代次数,正则参数]
    val model = ALS.train(trainData, rank, iterations, lambda)



    //使用模型预测
    //5.计算用户推荐矩阵(每一个用户对所有的电影)
    //需要构造一个usersProducts  RDD[(Int,Int)]
    val userMovies = userRDD.cartesian(movieRDD)
    //模型预测[Rating是它自带的样例类][这是所有的都已经进行了评分]
    val preRatings: RDD[Rating] = model.predict(userMovies)
    //用户推荐矩阵
   /* val userRecs = preRatings
      //把评分值大于零的过滤掉dm
      .filter(_.rating > 0)
      .map(rating => (rating.user, (rating.product, rating.rating)))
      .groupByKey()
      .map {
        case (uid, recs) => UserRecs(uid, recs.toList.sortWith(_._2 > _._2).take(USER_MAX_RECOMMENDATION).map(x => Recommendation(x._1, x._2)))
      }.toDF()*/





    //分解出商品的矩阵
    val features: RDD[(Int, Array[Double])] = model.productFeatures
    //6.电影相似度矩阵[电影的id  和电影的评分]
    val moviesFeatures = features.map { case (mid, features) => {
      //把特征换成向量的
      (mid, new DoubleMatrix(features))
    }
    }
    //求电影之间的相似度
    val movieRecs = moviesFeatures.cartesian(moviesFeatures).filter {
      case (a, b) => a._1 != b._1
    }.map {
      case (a, b) => {
        //电影A和电影B之间的相似度
        val simScore = consinSim(a._2, b._2)
        //simScore  值 必须大于0.5 才是有意义的
        (a._1, (b._1, simScore))
      }
    } //过滤的条件只管重要
      .filter(_._2._2.toString().toDouble > 0.8)
      .groupByKey().map { case (mid, items) =>
      MovieRecs(mid, items.toList.map(x => Recommendation(x._1, x._2)))
    }.toDF()

   /* //用户推荐矩阵写入mongodb
    userRecs.write
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_USER_RECS_COLLECTION)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()*/
    //电影相似度矩阵写入mongodb中
    movieRecs.write.
      option("uri", mongoConfig.uri)
      .option("collection", MONGODB_MOVIE_RECS_COLLECTION)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()
    //7.关闭spark
    spark.close()


  }

}

/**
  * MongoDB 配置对象
  *
  * @param uri MongoDB连接地址
  * @param db  操作的MongoDB数据库
  */
case class MongoConfig(val uri: String, val db: String)

/**
  * Rating Class 电影的评分类
  *
  * @param uid       用户的ID
  * @param mid       电影的ID
  * @param score     用户为该电影的评分
  * @param timestamp 用户为该电影评分的时间
  */
case class MoviesRating(val uid: Int, val mid: Int, val score: Double, val timestamp: Int)

/**
  * Movie Class 电影类
  *
  * @param mid       电影的ID
  * @param name      电影的名称
  * @param descri    电影的描述
  * @param timelong  电影的时长
  * @param issue     电影的发行时间
  * @param shoot     电影的拍摄时间
  * @param language  电影的语言
  * @param genres    电影的类别
  * @param actors    电影的演员
  * @param directors 电影的导演
  */
case class Movie(val mid: Int, val name: String, val descri: String, val timelong: String, val issue: String, val shoot: String, val language: String, val genres: String, val actors: String, val directors: String)

/**
  * 用户的推荐 [用户的id,推荐内容]
  *
  * @param uid
  * @param recs
  */
case class UserRecs(uid: Int, recs: Seq[Recommendation])

/**
  * 推荐项目
  *
  * @param rid 电影ID
  * @param r   推荐分数
  */
case class Recommendation(rid: Int, r: Double)

/**
  * 电影推荐矩阵里面封装的内容
  *
  * @param mid
  * @param recommendations
  */
case class MovieRecs(mid: Int, recommendations: List[Recommendation])

